package com.lyrcsoft.webflux_01.reactive;

import java.util.concurrent.*;

public class ReactiveStreamDemo {
    public static void main(String[] args) {
        // 1.发布者
        SubmissionPublisher<String> publisher = new SubmissionPublisher<String>();
        // 2.设置中间节点的处理器
        ReactiveProcessor processor = new ReactiveProcessor();
        // 3.由发布者与处理之间建立订阅关系
        publisher.subscribe(processor);
        // 4.订阅者
        Flow.Subscriber<String> subscriber = new Flow.Subscriber<String>() {
            // 将关系那出来，后续可能需要用到
            Flow.Subscription subscription;

            @Override
            public void onSubscribe(Flow.Subscription subscription) { // 入参 订阅协议
                System.out.println("subscriber:订阅关系");
                this.subscription = subscription;
                this.subscription.request(1);
            }

            @Override
            public void onNext(String item) {
                System.out.println("接收数据：" + item);
                // 每次接收数据 都走onNext，发100接100
                this.subscription.request(1);
            }

            @Override
            public void onError(Throwable throwable) {
                System.out.println("onError：" + throwable.getMessage());
            }

            @Override
            public void onComplete() {
                System.out.println("结束：onComplete");
            }
        };
        // 建立订阅关系(会生成Subscription 契约)
        processor.subscribe(subscriber);
        // 使用ExecutorService来在另一个线程中发布数据
        ExecutorService executor = Executors.newSingleThreadExecutor();
        Future<?> submit = executor.submit(() -> {
            publisher.submit("WebFlux");
            publisher.submit("Reactive Programming");
        });
        try {
            Object o = submit.get();
            // 完成后关闭发布者
            publisher.close();
            System.out.println("关闭发布者");
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (ExecutionException e) {
            e.printStackTrace();
        }

        //executor.shutdown();
    }
}
